package com.katesoft.scale4j.rttp.jobs.tasklets;

import com.hazelcast.core.IMap;
import com.katesoft.scale4j.rttp.client.IClusterInstanceAware.ClusterInstanceAwareBean;
import com.katesoft.scale4j.rttp.client.ISpringContextAccessor;
import com.katesoft.scale4j.common.lang.RuntimeUtility;
import com.katesoft.scale4j.common.services.IBeanNameReferences;
import com.katesoft.scale4j.log.Logger;
import com.katesoft.scale4j.log.LogFactory;
import com.katesoft.scale4j.persistent.client.LocalHibernateTemplate;
import com.katesoft.scale4j.persistent.model.unified.AbstractPersistentEntity;
import org.hibernate.HibernateException;
import org.hibernate.Session;
import org.hibernate.cache.CacheKey;
import org.hibernate.cache.entry.CacheEntry;
import org.hibernate.event.EventSource;
import org.hibernate.persister.entity.AbstractEntityPersister;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.orm.hibernate3.HibernateCallback;

import java.sql.SQLException;
import java.util.Iterator;
import java.util.Map.Entry;

/**
 * this class will clean cache if necessary.
 *
 * @author kate2007
 */
public class CacheCleanTasklet extends ClusterInstanceAwareBean implements Tasklet
{
    public static final String ENTITY_CLASS = "rttp.cache.clean.entity.class";
    private final Logger log = LogFactory.getLogger(getClass());

    @Override
    public RepeatStatus execute(StepContribution contribution,
                                ChunkContext chunkContext) throws Exception
    {
        final String region = chunkContext.getStepContext().getStepExecution().getJobParameters().getString(ENTITY_CLASS);
        if (getInstance().getLifecycleService().isRunning()) {
            final IMap<Object, Object> map = getInstance().getMap(region);
            final LocalHibernateTemplate template =
                ISpringContextAccessor.ACCESSOR.getBean(IBeanNameReferences.HIBERNATE_TEMPLATE, LocalHibernateTemplate.class);
            template.execute(new HibernateCallback<Object>()
            {
                @Override
                public Object doInHibernate(Session session) throws HibernateException, SQLException
                {
                    for (Iterator<Entry<Object, Object>> iterator = map.entrySet().iterator(); iterator.hasNext();) {
                        Entry<Object, Object> entry = iterator.next();
                        CacheEntry cacheEntry = (CacheEntry) entry.getValue();
                        CacheKey cacheKey = (CacheKey) entry.getKey();
                        AbstractPersistentEntity obj = (AbstractPersistentEntity) RuntimeUtility.newInstance(region);
                        AbstractEntityPersister p = ((AbstractEntityPersister) template.getSessionFactory().getClassMetadata(obj.getClass()));
                        cacheEntry.assemble(obj, cacheKey.getKey(), p, null, (EventSource) session);
                        if (obj.isCleanable()) { iterator.remove(); }
                    }
                    RuntimeUtility.gc();
                    return null;
                }
            });
            int oldSize = map.size();
            log.info("region[%s] cleaned ::: old_size = %s, new_size = %s", region, oldSize, map.size());
        }
        return RepeatStatus.FINISHED;
    }
}
